package com.haozhen.streamtransformation;

import com.haozhen.datasource.SelfSourceParallel;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.security.Principal;
import java.util.ArrayList;
import java.util.List;

/**
 * @author haozhen
 * @email haozh@ync1.com
 * @date 2021/2/24  2:25
 */
public class SplitDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> data1 = env.addSource(new SelfSourceParallel());

        SplitStream<String> split = data1.split(new OutputSelector<String>() {
            @Override
            public Iterable<String> select(String value) {
                int i = Integer.parseInt(value);
                List<String> output = new ArrayList<>();
                if (i % 2 == 0) {
                    output.add("even");
                } else {
                    output.add("odd");
                }
                return output;
            }
        });
        DataStream<String> even = split.select("even");
        DataStream<String> odd = split.select("odd");
        System.out.println("even");
        even.print();
//        System.out.println("odd");
//        odd.print();
        env.execute();
    }
}
